--------------------------sql任务说明----------------------------------
-- @author :liuxiaojie05
-- @descrbe :简历下载数据 source=kafka b2c ods --->dwd
-- @udf jar包 : bigdata_realtime_flink_sql_udf-1.0-testlxj-jar-with-dependencies.jar
-- 下游使用：dmp、olap、push

--------------------------sql任务说明----------------------------------

-- 加载函数
CREATE FUNCTION ZpDWDResumeDownLoadV2Udtf as 'udtf.etl.dwd.ZpDWDResumeDownLoadV2Udtf';
create function SplitFunction AS 'udtf.SplitFunction';


-- source
create table ods_zp_delivery_downloadv2 (
log string
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_zhaopin_resumedown',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_resume_download_v2-YqlGU-consumer',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_zhaopin_resumedown-alZRg',
'format' = 'raw'
);

-- sink
create table dwd_zp_delivery_downloadv2 (
    euserid         string,
    confuseResumeId string,
    sex           string,
    age           string,
    educationName string,
    workyearsName string,
    downTime      string,
    consumeSourceCount string,
    euserType      string,
    downPlatform   string,
    downPayType string,
    cuserId     string,
    resumeId    string,
    productType string,
    zpbizid     string,
    rootUserId  string,
    seriesId string,
    sver     string,
    sid      string,
    slotid   string,
    e_time   string
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_dwd_zp_resume_download_v2',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin_dwd_zp_resume_download_v2-sql-prod_v1',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_resume_download_v2-YqlGU',
'format' = 'json'
);


-- sql-etl
insert into dwd_zp_delivery_downloadv2
select pp.*
from (
    select SPLIT_INDEX(sl.regex_str, ' - ', 1) as message
    from (
        select regex_str
        from ods_zp_delivery_downloadv2,
        LATERAL TABLE(SplitFunction(log,'\n'))
    ) as sl
    where SPLIT_INDEX(sl.regex_str, ' - ', 1) is not null
    and CHAR_LENGTH(SPLIT_INDEX(sl.regex_str, ' - ', 1)) > 1
) as tmp,
LATERAL TABLE(ZpDWDResumeDownLoadV2Udtf(message)) as pp
-- where  char_length(pp.euserid) > 1000
;